media_pp\elements\sink\renderer\windows/
wasapi_renderer.rs

1use std::{ffi::c_void, ptr, sync::Arc, thread, time::Duration};
2
3use crate::pp_log::{PpLog, pp_error, pp_info};
4use ffmpeg_next::{self as ffmpeg, Rescale, Rounding};
5use thiserror::Error as ThisError;
6use windows::Win32::{
7    Media::Audio::{
8        AUDCLNT_E_DEVICE_INVALIDATED, AUDCLNT_SHAREMODE_SHARED, IAudioClient, IAudioClock,
9        IAudioRenderClient,
10    },
11    System::Com::{CLSCTX_ALL, CoTaskMemFree},
12};
13
14use crate::{
15    buffer::MediaBuffer,
16    control::ControlMsg,
17    element::{Element, ElementType, Sink, element_pp_log},
18    elements::{AudioFormat, WasapiDevice, WasapiDeviceKind},
19    error::Result,
20    platform::windows::wasapi::{
21        ComApartment, list_devices as enumerate_wasapi_devices, open_device, resolve_mix_format,
22    },
23    playback_clock::{AudioMasterRegistration, PlaybackClock, PlaybackClockError},
24    time::{MediaTimestamp, TimeBase},
25};
26
27const BUFFER_DURATION_100NS: i64 = 100 * 10_000;
28const POLL_INTERVAL: Duration = Duration::from_millis(2);
29
30#[derive(Debug, Clone)]
31pub struct WasapiRendererOptions {
32    pub device: WasapiDevice,
33}
34
35#[derive(Debug, ThisError)]
36pub enum WasapiRendererError {
37    #[error("windows error: {0}")]
38    Windows(#[from] windows::core::Error),
39
40    #[error("AUDCLNT_E_DEVICE_INVALIDATED — audio device needs to be reopened")]
41    DeviceInvalidated,
42
43    #[error("WasapiRenderer requires a Render endpoint, got {0:?}")]
44    NotRenderDevice(WasapiDeviceKind),
45
46    #[error("unsupported WASAPI mix format: format_tag={format_tag}, bits_per_sample={bits}")]
47    UnsupportedMixFormat { format_tag: u32, bits: u16 },
48
49    #[error(
50        "audio format mismatch: expected {expected:?}, got {actual:?}; insert AudioResampler before WasapiRenderer"
51    )]
52    FormatMismatch {
53        expected: AudioFormat,
54        actual: AudioFormat,
55    },
56
57    #[error("audio frame buffer is shorter than its declared sample count")]
58    TruncatedFrame,
59
60    #[error("WasapiRenderer only renders decoded Audio frames, got a {0}")]
61    UnsupportedBuffer(&'static str),
62
63    #[error(transparent)]
64    PlaybackClock(#[from] PlaybackClockError),
65
66    #[error("cannot bind a playback clock after this audio endpoint has started")]
67    PlaybackClockBoundAfterStart,
68
69    #[error("this WasapiRenderer is already bound to a playback clock")]
70    PlaybackClockAlreadyBound,
71
72    #[error("audio frames need a PTS when WasapiRenderer is the playback-clock master")]
73    MissingPts,
74
75    #[error("WASAPI reported an invalid audio-clock frequency of {0}")]
76    InvalidClockFrequency(u64),
77}
78
79/// Terminal audio sink backed by a WASAPI shared-mode render endpoint.
80/// The endpoint's mix format is returned by [`WasapiRenderer::open`] so a
81/// caller can place an [`crate::elements::AudioResampler`] immediately
82/// before this sink. This element intentionally performs no hidden format
83/// conversion. Call [`WasapiRenderer::bind_playback_clock`] while wiring a
84/// fixed A/V pipeline to publish this endpoint's actual played-sample position
85/// as that pipeline's audio master. A branch attached to a running dynamic Tee
86/// uses [`WasapiRenderer::bind_playback_clock_deferred`] instead, so it cannot
87/// stall video before the first audio frame reaches the renderer.
88///
89/// Device-buffer backpressure is the playback clock: `consume` waits for
90/// enough WASAPI ring-buffer space to submit the whole input frame. Put a
91/// [`crate::queue::Queue`] immediately before this sink when its blocking
92/// must not hold up another branch.
93pub struct WasapiRenderer {
94    pp_log: PpLog,
95    name: Arc<str>,
96    audio_client: IAudioClient,
97    audio_clock: IAudioClock,
98    audio_clock_frequency: u64,
99    render_client: IAudioRenderClient,
100    format: AudioFormat,
101    buffer_frames: u32,
102    running: bool,
103    paused: bool,
104    clock_binding: PlaybackClockBinding,
105    timeline: Option<DeviceTimeline>,
106}
107
108enum PlaybackClockBinding {
109    Unbound,
110    Deferred(Arc<PlaybackClock>),
111    Registered(AudioMasterRegistration),
112}
113
114impl PlaybackClockBinding {
115    fn is_bound(&self) -> bool {
116        !matches!(self, Self::Unbound)
117    }
118
119    fn registration(&self) -> Option<&AudioMasterRegistration> {
120        match self {
121            Self::Registered(master) => Some(master),
122            Self::Unbound | Self::Deferred(_) => None,
123        }
124    }
125
126    fn ensure_registered(&mut self) -> std::result::Result<(), PlaybackClockError> {
127        let registration = match self {
128            Self::Deferred(playback_clock) => Some(playback_clock.register_audio_master()?),
129            Self::Unbound | Self::Registered(_) => None,
130        };
131        if let Some(registration) = registration {
132            *self = Self::Registered(registration);
133        }
134        Ok(())
135    }
136}
137
138struct DeviceTimeline {
139    device_origin: u64,
140    media_origin_ns: i64,
141    submitted_until_ns: i64,
142}
143
144// SAFETY: WASAPI client interfaces are free-threaded. Every method that
145// touches them requires `&mut self`, and each calling thread joins a COM
146// apartment for the duration of the call via `ComApartment`.
147unsafe impl Send for WasapiRenderer {}
148
149impl WasapiRenderer {
150    pub fn list_devices() -> std::result::Result<Vec<WasapiDevice>, WasapiRendererError> {
151        Ok(enumerate_wasapi_devices(Some(WasapiDeviceKind::Render))?)
152    }
153
154    pub fn open(
155        name: impl Into<String>,
156        options: WasapiRendererOptions,
157    ) -> std::result::Result<(Self, AudioFormat), WasapiRendererError> {
158        if options.device.kind != WasapiDeviceKind::Render {
159            return Err(WasapiRendererError::NotRenderDevice(options.device.kind));
160        }
161
162        let _apartment = ComApartment::new()?;
163        let name: Arc<str> = name.into().into();
164        let pp_log = element_pp_log(ElementType::WasapiRenderer, &name, None);
165        let device = open_device(&options.device.id)?;
166        let audio_client: IAudioClient = unsafe { device.Activate(CLSCTX_ALL, None)? };
167        let mix_format = unsafe { audio_client.GetMixFormat()? };
168        let format = resolve_mix_format(mix_format).map_err(|error| {
169            WasapiRendererError::UnsupportedMixFormat {
170                format_tag: error.format_tag,
171                bits: error.bits,
172            }
173        })?;
174        let initialize_result = unsafe {
175            audio_client.Initialize(
176                AUDCLNT_SHAREMODE_SHARED,
177                0,
178                BUFFER_DURATION_100NS,
179                0,
180                mix_format,
181                None,
182            )
183        };
184        unsafe { CoTaskMemFree(Some(mix_format as *const c_void)) };
185        initialize_result?;
186
187        let render_client: IAudioRenderClient = unsafe { audio_client.GetService()? };
188        let audio_clock: IAudioClock = unsafe { audio_client.GetService()? };
189        let audio_clock_frequency = unsafe { audio_clock.GetFrequency()? };
190        if audio_clock_frequency == 0 {
191            return Err(WasapiRendererError::InvalidClockFrequency(
192                audio_clock_frequency,
193            ));
194        }
195        let buffer_frames = unsafe { audio_client.GetBufferSize()? };
196        pp_info!(
197            pp_log: &pp_log,
198            "opened: device={:?}, {}Hz, {} channel(s), format={:?}, buffer_frames={buffer_frames}",
199            options.device.name,
200            format.sample_rate,
201            format.channels,
202            format.sample_format
203        );
204
205        Ok((
206            Self {
207                name,
208                pp_log,
209                audio_client,
210                audio_clock,
211                audio_clock_frequency,
212                render_client,
213                format,
214                buffer_frames,
215                running: false,
216                paused: false,
217                clock_binding: PlaybackClockBinding::Unbound,
218                timeline: None,
219            },
220            format,
221        ))
222    }
223
224    pub fn format(&self) -> AudioFormat {
225        self.format
226    }
227
228    /// Makes this endpoint the pipeline's exclusive audio playback master.
229    /// Call during the wiring closure, before boxing the renderer into its
230    /// terminal branch.
231    pub fn bind_playback_clock(
232        &mut self,
233        playback_clock: Arc<PlaybackClock>,
234    ) -> std::result::Result<(), WasapiRendererError> {
235        if self.clock_binding.is_bound() {
236            return Err(WasapiRendererError::PlaybackClockAlreadyBound);
237        }
238        if self.running || self.timeline.is_some() {
239            return Err(WasapiRendererError::PlaybackClockBoundAfterStart);
240        }
241        let master = playback_clock.register_audio_master()?;
242        self.clock_binding = PlaybackClockBinding::Registered(master);
243        Ok(())
244    }
245
246    /// Binds a dynamically attached endpoint without claiming the audio-master
247    /// slot until its first non-empty audio frame arrives.
248    ///
249    /// This avoids a priming deadlock when an upstream demuxer can block on a
250    /// full video queue before reaching the first packet for the newly attached
251    /// audio branch. Unlike [`Self::bind_playback_clock`], an exclusive-master
252    /// conflict is therefore returned from that first [`Sink::consume`] call.
253    pub fn bind_playback_clock_deferred(
254        &mut self,
255        playback_clock: Arc<PlaybackClock>,
256    ) -> std::result::Result<(), WasapiRendererError> {
257        if self.clock_binding.is_bound() {
258            return Err(WasapiRendererError::PlaybackClockAlreadyBound);
259        }
260        if self.running || self.timeline.is_some() {
261            return Err(WasapiRendererError::PlaybackClockBoundAfterStart);
262        }
263        self.clock_binding = PlaybackClockBinding::Deferred(playback_clock);
264        Ok(())
265    }
266
267    fn ensure_playback_master(&mut self) -> Result<()> {
268        self.clock_binding
269            .ensure_registered()
270            .map_err(WasapiRendererError::from)?;
271        Ok(())
272    }
273
274    fn classify_error(&self, error: windows::core::Error) -> WasapiRendererError {
275        if error.code() == AUDCLNT_E_DEVICE_INVALIDATED {
276            WasapiRendererError::DeviceInvalidated
277        } else {
278            WasapiRendererError::Windows(error)
279        }
280    }
281
282    fn start(&mut self) -> Result<()> {
283        if !self.running {
284            unsafe { self.audio_client.Start() }.map_err(|error| self.classify_error(error))?;
285            self.running = true;
286        }
287        Ok(())
288    }
289
290    fn stop_and_reset(&mut self) -> Result<()> {
291        if self.running {
292            unsafe { self.audio_client.Stop() }.map_err(|error| self.classify_error(error))?;
293        }
294        self.running = false;
295        self.publish_device_position(false)?;
296        unsafe { self.audio_client.Reset() }.map_err(|error| self.classify_error(error))?;
297        self.timeline = None;
298        Ok(())
299    }
300
301    fn device_position(&self) -> std::result::Result<u64, WasapiRendererError> {
302        let mut position = 0;
303        unsafe { self.audio_clock.GetPosition(&mut position, None) }
304            .map_err(|error| self.classify_error(error))?;
305        Ok(position)
306    }
307
308    fn publish_device_position(&self, running: bool) -> Result<()> {
309        let (Some(master), Some(timeline)) = (self.clock_binding.registration(), &self.timeline)
310        else {
311            return Ok(());
312        };
313        let position = self.device_position()?;
314        let device_delta = position.saturating_sub(timeline.device_origin);
315        let elapsed_ns = ((u128::from(device_delta) * 1_000_000_000u128)
316            / u128::from(self.audio_clock_frequency))
317        .min(i64::MAX as u128) as i64;
318        master
319            .publish(
320                timeline.media_origin_ns.saturating_add(elapsed_ns),
321                timeline.submitted_until_ns,
322                running,
323            )
324            .map_err(WasapiRendererError::from)?;
325        Ok(())
326    }
327
328    fn audio_pts_ns(&self, frame: &ffmpeg::frame::Audio) -> Result<i64> {
329        let pts = frame.pts().ok_or(WasapiRendererError::MissingPts)?;
330        let source =
331            TimeBase::new_unchecked(ffmpeg::Rational::new(1, self.format.sample_rate as i32));
332        let nanos = TimeBase::new_unchecked(ffmpeg::Rational::new(1, 1_000_000_000));
333        Ok(MediaTimestamp::new_unchecked(pts, source).rescale(nanos))
334    }
335
336    fn sample_offset_ns(&self, samples: usize) -> i64 {
337        (samples as i64).rescale(
338            ffmpeg::Rational::new(1, self.format.sample_rate as i32),
339            ffmpeg::Rational::new(1, 1_000_000_000),
340        )
341    }
342
343    fn render(&mut self, frame: &ffmpeg::frame::Audio) -> Result<()> {
344        let bytes = validate_frame(self.format, frame)?;
345        if frame.samples() == 0 || self.paused {
346            return Ok(());
347        }
348        self.ensure_playback_master()?;
349
350        let bytes_per_frame = self.format.sample_format.bytes() * self.format.channels as usize;
351        let frame_pts_ns = if self.clock_binding.registration().is_some() {
352            Some(self.audio_pts_ns(frame)?)
353        } else {
354            None
355        };
356        let mut frame_offset = 0usize;
357        if let (Some(master), Some(frame_pts_ns)) =
358            (self.clock_binding.registration(), frame_pts_ns)
359            && let Some(target_ns) = master
360                .priming_target_ns()
361                .map_err(WasapiRendererError::from)?
362        {
363            let delta_ns = target_ns.saturating_sub(frame_pts_ns);
364            if delta_ns > 0 {
365                frame_offset =
366                    priming_trim_samples(frame_pts_ns, target_ns, self.format.sample_rate);
367                if frame_offset >= frame.samples() {
368                    return Ok(());
369                }
370            }
371        }
372
373        while frame_offset < frame.samples() {
374            let padding = unsafe { self.audio_client.GetCurrentPadding() }
375                .map_err(|error| self.classify_error(error))?;
376            // IAudioClock keeps advancing through an endpoint underrun.
377            // If the previous submitted range has fully drained, map the
378            // next real sample to the current device position instead of
379            // counting the intervening silence as media time.
380            let rebase_timeline = padding == 0 && self.running && self.timeline.is_some();
381            let available = self.buffer_frames.saturating_sub(padding) as usize;
382            if available == 0 {
383                self.start()?;
384                self.publish_device_position(true)?;
385                thread::sleep(POLL_INTERVAL);
386                continue;
387            }
388
389            let take = available.min(frame.samples() - frame_offset);
390            let destination = unsafe { self.render_client.GetBuffer(take as u32) }
391                .map_err(|error| self.classify_error(error))?;
392            let byte_offset = frame_offset * bytes_per_frame;
393            let byte_count = take * bytes_per_frame;
394            unsafe {
395                ptr::copy_nonoverlapping(
396                    bytes[byte_offset..byte_offset + byte_count].as_ptr(),
397                    destination,
398                    byte_count,
399                );
400            }
401            unsafe { self.render_client.ReleaseBuffer(take as u32, 0) }
402                .map_err(|error| self.classify_error(error))?;
403            if let Some(frame_pts_ns) = frame_pts_ns {
404                let submitted_until_ns = frame_pts_ns
405                    .saturating_add(self.sample_offset_ns(frame_offset.saturating_add(take)));
406                if !rebase_timeline && let Some(timeline) = &mut self.timeline {
407                    timeline.submitted_until_ns = submitted_until_ns;
408                } else {
409                    self.timeline = Some(DeviceTimeline {
410                        device_origin: self.device_position()?,
411                        media_origin_ns: frame_pts_ns
412                            .saturating_add(self.sample_offset_ns(frame_offset)),
413                        submitted_until_ns,
414                    });
415                }
416            }
417            frame_offset += take;
418            self.start()?;
419            self.publish_device_position(true)?;
420        }
421        Ok(())
422    }
423
424    fn drain(&mut self) -> Result<()> {
425        let padding = unsafe { self.audio_client.GetCurrentPadding() }
426            .map_err(|error| self.classify_error(error))?;
427        if padding > 0 {
428            self.start()?;
429        }
430        loop {
431            let padding = unsafe { self.audio_client.GetCurrentPadding() }
432                .map_err(|error| self.classify_error(error))?;
433            if padding == 0 {
434                break;
435            }
436            self.publish_device_position(true)?;
437            thread::sleep(POLL_INTERVAL);
438        }
439        self.publish_device_position(false)?;
440        let final_position = self
441            .timeline
442            .as_ref()
443            .map(|timeline| timeline.submitted_until_ns);
444        self.stop_and_reset()?;
445        if let (Some(master), Some(final_position)) =
446            (self.clock_binding.registration(), final_position)
447        {
448            master
449                .finish(final_position)
450                .map_err(WasapiRendererError::from)?;
451        }
452        Ok(())
453    }
454}
455
456fn validate_frame(
457    expected: AudioFormat,
458    frame: &ffmpeg::frame::Audio,
459) -> std::result::Result<&[u8], WasapiRendererError> {
460    let actual = AudioFormat::new(frame.format(), frame.rate(), frame.channels());
461    if actual != expected {
462        return Err(WasapiRendererError::FormatMismatch { expected, actual });
463    }
464    let tight_bytes = frame
465        .samples()
466        .saturating_mul(expected.channels as usize)
467        .saturating_mul(expected.sample_format.bytes());
468    frame
469        .data(0)
470        .get(..tight_bytes)
471        .ok_or(WasapiRendererError::TruncatedFrame)
472}
473
474fn priming_trim_samples(frame_pts_ns: i64, target_ns: i64, sample_rate: u32) -> usize {
475    target_ns
476        .saturating_sub(frame_pts_ns)
477        .max(0)
478        .rescale_with(
479            ffmpeg::Rational::new(1, 1_000_000_000),
480            ffmpeg::Rational::new(1, sample_rate as i32),
481            Rounding::Up,
482        )
483        .max(0) as usize
484}
485
486impl Element for WasapiRenderer {
487    fn name(&self) -> Arc<str> {
488        self.name.clone()
489    }
490
491    fn element_type(&self) -> ElementType {
492        ElementType::WasapiRenderer
493    }
494
495    fn pp_log(&self) -> &PpLog {
496        &self.pp_log
497    }
498
499    fn pp_log_mut(&mut self) -> &mut PpLog {
500        &mut self.pp_log
501    }
502}
503
504impl Sink for WasapiRenderer {
505    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
506        let _apartment = ComApartment::new().map_err(WasapiRendererError::from)?;
507        match buf {
508            MediaBuffer::Audio(frame) => self
509                .render(&frame)
510                .inspect_err(|error| pp_error!(self, "render failed: {error}")),
511            MediaBuffer::Eos => self
512                .drain()
513                .inspect_err(|error| pp_error!(self, "drain failed: {error}")),
514            MediaBuffer::Packet(_) => Err(WasapiRendererError::UnsupportedBuffer("Packet").into()),
515            MediaBuffer::Video(_) => Err(WasapiRendererError::UnsupportedBuffer("Video").into()),
516        }
517    }
518
519    fn control(&mut self, msg: ControlMsg) -> Result<()> {
520        let _apartment = ComApartment::new().map_err(WasapiRendererError::from)?;
521        match msg {
522            ControlMsg::Pause => {
523                if self.running {
524                    unsafe { self.audio_client.Stop() }
525                        .map_err(|error| self.classify_error(error))?;
526                    self.running = false;
527                }
528                self.publish_device_position(false)?;
529                self.paused = true;
530            }
531            ControlMsg::Resume => {
532                self.paused = false;
533                let padding = unsafe { self.audio_client.GetCurrentPadding() }
534                    .map_err(|error| self.classify_error(error))?;
535                if padding > 0 {
536                    self.start()?;
537                    self.publish_device_position(true)?;
538                }
539            }
540            ControlMsg::Stop => {
541                self.paused = false;
542                self.stop_and_reset()?;
543            }
544            ControlMsg::Seek(_) => {
545                if self.running {
546                    unsafe { self.audio_client.Stop() }
547                        .map_err(|error| self.classify_error(error))?;
548                }
549                self.running = false;
550                self.paused = false;
551                unsafe { self.audio_client.Reset() }.map_err(|error| self.classify_error(error))?;
552                self.timeline = None;
553                if let Some(master) = self.clock_binding.registration() {
554                    master.reset_for_seek().map_err(WasapiRendererError::from)?;
555                }
556            }
557        }
558        Ok(())
559    }
560}
561
562impl Drop for WasapiRenderer {
563    fn drop(&mut self) {
564        let Ok(_apartment) = ComApartment::new() else {
565            return;
566        };
567        if self.running {
568            let _ = unsafe { self.audio_client.Stop() };
569            self.running = false;
570        }
571        // A dynamically detached renderer must hand the last actually played
572        // position back to PlaybackClock before its master registration drops.
573        // Otherwise video can resume wall-clock pacing from the last periodic
574        // update, a few milliseconds behind the audible handoff point.
575        let _ = self.publish_device_position(false);
576        let _ = unsafe { self.audio_client.Reset() };
577        self.timeline = None;
578    }
579}
580
581#[cfg(test)]
582mod tests {
583    use ffmpeg::format::sample::Type;
584
585    use super::*;
586    use crate::{clock::Clock, playback_clock::PlaybackMaster};
587
588    fn frame(format: AudioFormat, samples: usize) -> ffmpeg::frame::Audio {
589        let mut frame =
590            ffmpeg::frame::Audio::new(format.sample_format, samples, format.channel_layout());
591        frame.set_rate(format.sample_rate);
592        frame.data_mut(0).fill(0);
593        frame
594    }
595
596    #[test]
597    fn binding_does_not_claim_the_clock_until_audio_can_prime_it() {
598        let playback = Arc::new(PlaybackClock::new(Arc::new(Clock::new())));
599        playback.ensure_wall_origin(1_000);
600        let mut binding = PlaybackClockBinding::Deferred(playback.clone());
601
602        assert_eq!(playback.master(), PlaybackMaster::Wall);
603        binding.ensure_registered().unwrap();
604        assert!(matches!(binding, PlaybackClockBinding::Registered(_)));
605        assert_eq!(playback.master(), PlaybackMaster::AudioPriming);
606    }
607
608    #[test]
609    fn failed_deferred_registration_keeps_the_deferred_state() {
610        let playback = Arc::new(PlaybackClock::new(Arc::new(Clock::new())));
611        let _existing_master = playback.register_audio_master().unwrap();
612        let mut binding = PlaybackClockBinding::Deferred(playback);
613
614        assert!(matches!(
615            binding.ensure_registered(),
616            Err(PlaybackClockError::AudioMasterAlreadyRegistered)
617        ));
618        assert!(matches!(binding, PlaybackClockBinding::Deferred(_)));
619    }
620
621    #[test]
622    fn validates_the_exact_device_mix_format() {
623        let expected = AudioFormat::new(ffmpeg::format::Sample::F32(Type::Packed), 48_000, 2);
624        let frame = frame(expected, 480);
625        assert_eq!(validate_frame(expected, &frame).unwrap().len(), 480 * 2 * 4);
626    }
627
628    #[test]
629    fn rejects_audio_that_skipped_the_required_resampler() {
630        let expected = AudioFormat::new(ffmpeg::format::Sample::F32(Type::Packed), 48_000, 2);
631        let actual = AudioFormat::new(ffmpeg::format::Sample::I16(Type::Packed), 44_100, 1);
632        let error = validate_frame(expected, &frame(actual, 441)).unwrap_err();
633        assert!(matches!(
634            error,
635            WasapiRendererError::FormatMismatch {
636                expected: error_expected,
637                actual: error_actual,
638            } if error_expected == expected && error_actual == actual
639        ));
640    }
641
642    #[test]
643    fn priming_trim_rounds_forward_to_the_first_sample_not_before_wall_position() {
644        assert_eq!(priming_trim_samples(0, 10_000_001, 48_000), 481);
645        assert_eq!(priming_trim_samples(20_000_000, 10_000_000, 48_000), 0);
646    }
647}
media_pp\elements\sink\renderer\windows/wasapi_renderer.rs

media_pp\elements\sink\renderer\windows/
wasapi_renderer.rs
